/* * Copyright © 2023 https://www.cestc.cn/ All rights reserved. */
package com.ffz.calcitesql.demo.pulsar;

import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.SubscriptionInitialPosition;
import org.apache.pulsar.client.api.SubscriptionType;

import java.util.concurrent.TimeUnit;

/**
 * 请对类注释说明.
 *
 * @author fengfazheng
 * @since 2024/1/31 13:52
 */
public class ConsumerDemo1 {

    //Pulsar集群中broker的serviceurl
    private static final String brokerServiceurl = "pulsar://10.255.158.50:6650";
    //需要订阅的topic name
    private static final String topicName = "persistent://Flink/source_demo/online_sink";
    //订阅名
    private static final String subscriptionName = "online-subscription";

    public static void main(String[] args) throws PulsarClientException {
        //构造Pulsar client
        PulsarClient client = PulsarClient.builder()
                .serviceUrl(brokerServiceurl)
                .build();

        //创建consumer
        Consumer consumer = client.newConsumer()
                .topic(topicName)
                .subscriptionName(subscriptionName)
                .subscriptionType(SubscriptionType.Exclusive)//指定消费模式，包含：Exclusive，Failover，Shared，Key_Shared。默认Exclusive模式
                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)//指定从哪里开始消费，还有Latest，valueof可选，默认Latest
                .negativeAckRedeliveryDelay(60, TimeUnit.SECONDS)//指定消费失败后延迟多久broker重新发送消息给consumer，默认60s
                .subscribe();

        //消费消息
        while (true) {
            Message message = consumer.receive();

            try {
                System.out.println("******************"+ new String(message.getData())+"********************");
//                System.out.printf(
//                        "Message received: time=%s, value=%s, topic=%s, id=%s,message =%s%n",
//                        new SimpleDateFormat("yyyy-MM-dd HH-mm-ss").format(message.getPublishTime()),
//                        message.getValue(),
//                        message.getTopicName(),
//                        message.getMessageId().toString(),new String(message.getData()));
//                System.out.printf("Message received: %s%n", new String(message.getData()));
//                consumer.acknowledge(message);
            } catch (Exception e) {
                e.printStackTrace();
                consumer.negativeAcknowledge(message);
            }

        }
    }


}
